package com.xqcx.libra.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.xqcx.libra.config.CanalConnectorConfig;
import com.xqcx.libra.util.StorageMessage;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;


@Slf4j
@Component
public class ClusterCanalClient {

    @Autowired
    protected StorageMessage storage;
    @Autowired
    private CanalConnector canalConnector;
    @Value("${destination}")
    private String destination ;
    @Value("${batchSize}")
    private int batchSize ;

    protected Thread thread = null;
    protected volatile boolean running = false;

    public ClusterCanalClient() {
    }

    public ClusterCanalClient(String destination, CanalConnector connector, StorageMessage storage) {
        this.destination = destination;
        this.canalConnector = connector;
        this.storage = storage;
    }

    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
        public void uncaughtException(Thread t, Throwable e) {
            log.error("parse events has an error", e);
        }
    };

    @PostConstruct
    protected void start() {
        Assert.notNull(canalConnector, "connector is null");
        thread = new Thread(new Runnable() {
            public void run() {
                process();
            }
        });
        thread.setUncaughtExceptionHandler(handler);
        running = true;
        thread.start();
    }

    @PreDestroy
    protected void stop() {
        if (!running) {
            return;
        }
        running = false;
        if (thread != null) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                // ignore
            }
        }
        MDC.remove("destination");
    }

    protected void process() {
        while (running) {
            try {
                MDC.put("destination", destination);
                canalConnector.connect();
                canalConnector.subscribe();
                while (running) {
                    // 获取指定数量的数据
                    Message message = canalConnector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(1000);
                        } catch (Exception e) {
                        }
                    } else {
                        storage.distributeMessage(message);
                    }
                    // 提交确认
                    canalConnector.ack(batchId);
                    // 处理失败, 回滚数据
                    // connector.rollback(batchId);
                }
            } catch (Exception e) {
                log.error("process error!", e);
            } finally {
                canalConnector.disconnect();
                MDC.remove("destination");
            }
        }
    }

}
